package interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;

import java.util.HashMap;

//自定义source
//接收数据（这里是for循环自己造数据）
//将数据封装为事件
//将事件发送到channel
public class MySource extends AbstractSource implements Configurable, PollableSource {

    //定义全局的前缀和后缀
    private String prefix;
    private String subfix;

    @Override
    public void configure(Context context) {
        //读取配置信息给前后缀赋值
       prefix = context.getString("prefix");
       subfix = context.getString("subfix","atguiguzmy");
    }


    @Override
    public Status process() throws EventDeliveryException {
        Status status = null;
        //造数据
        try {
            for (int i = 0; i < 5; i++) {
                //构建事件对象
                SimpleEvent event = new SimpleEvent();
                //给事件设置具体值
                event.setBody((prefix + "--" + i + "--" + subfix).getBytes());
                //将事件发送给channel
                getChannelProcessor().processEvent(event);
                status=Status.READY;
            }
        } catch (Exception e) {
            e.printStackTrace();
            status=Status.BACKOFF;
        }

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return status;
    }

    @Override
    public long getBackOffSleepIncrement() {
        return 0;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
        return 0;
    }


}
